/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.jms;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.Hashtable;

import javax.jms.JMSException;
import javax.jms.Message;

import cgl.narada.protocol.Protocol;
/** 
    This is the class which ties up the topic, Message, deliveryMode, priority
    and timeToLive as a single entity 
    @author Shrideep Pallickara
    $Revision$
    $Date$
*/

public class JmsEvent implements JmsDebugFlags {
  
  /** The topic to which this message was published */
  private String topic;
  
  /** The payload contained in the event */
  private Message message;
  
  /** The delivery mode specified by the publisher of this event */
  private int deliveryMode;
  
  /** The priority of the event */
  private int priority;
  
  /** The timeToLive parameter */
  private long timeToLive; 
  
  private int messageType=-20;
  
  private byte[] payloadBytes;
  
  public JmsEvent(String topic, Message message, 
		  int deliveryMode, int priority, long timeToLive) 
    throws JMSException {
    this.topic = topic;
    this.message = message;
    this.deliveryMode = deliveryMode;
    this.priority = priority;
    this.timeToLive = timeToLive;

     if (message.getJMSType().equals("TextMessage")) {
       messageType = JmsProtocolConstants.TextMessage_Type;
     } else if (message.getJMSType().equals("BytesMessage")) {
       messageType = JmsProtocolConstants.BytesMessage_Type;
     } else if (message.getJMSType().equals("ObjectMessage")) {
       messageType = JmsProtocolConstants.ObjectMessage_Type;
     } else if (message.getJMSType().equals("StreamMessage")) {
       messageType = JmsProtocolConstants.StreamMessage_Type;
     } else if (message.getJMSType().equals("MapMessage")) {
       messageType = JmsProtocolConstants.MapMessage_Type;
     }
  }
  
  
  public JmsEvent(byte[] jmsEventBytes) {
    ByteArrayInputStream baInputStream = 
      new ByteArrayInputStream(jmsEventBytes);
    DataInputStream din = new DataInputStream(baInputStream);
    
    try {
      din.readByte(); /** We can skip this since we know it is a JMS_EVENT*/
      
      topic = din.readUTF();

      messageType = din.readUnsignedByte();
      int payloadLength = din.readInt();
      payloadBytes = new byte[payloadLength];
      din.readFully(payloadBytes);
      if (messageType == JmsProtocolConstants.TextMessage_Type) {
	message = new JmsTextMessage(payloadBytes);
      } else if (messageType == JmsProtocolConstants.BytesMessage_Type) {
	message = new JmsBytesMessage(payloadBytes);
      } else if (messageType == JmsProtocolConstants.ObjectMessage_Type) {
	message = new JmsObjectMessage(payloadBytes);
      } else if (messageType == JmsProtocolConstants.StreamMessage_Type) {
	message = new JmsStreamMessage(payloadBytes);
      } else if (messageType == JmsProtocolConstants.MapMessage_Type) {
	message = new JmsMapMessage(payloadBytes);
      } else {
	System.out.println("JMSEvent:: Trying to unmarshall an unknown" + 
			   " data type");
      }
       
      deliveryMode = din.readInt();
      priority = din.readInt();
      timeToLive = din.readLong();
    } catch(Exception ioe) {
      System.out.println("JmsEvent:: Error during unmarshalling " + ioe);
    }
  }


  public JmsEvent(byte[] jmsEventBytes, boolean noPayloadReconstruct) {
    ByteArrayInputStream baInputStream = 
      new ByteArrayInputStream(jmsEventBytes);
    DataInputStream din = new DataInputStream(baInputStream);
    
    try {
      din.readByte(); /** We can skip this since we know it is a JMS_EVENT*/
      
      topic = din.readUTF();
      
      messageType = din.readUnsignedByte();
      int payloadLength = din.readInt();
      payloadBytes = new byte[payloadLength];
      din.readFully(payloadBytes);
      
      deliveryMode = din.readInt();
      priority = din.readInt();
      timeToLive = din.readLong();
    } catch(IOException ioe) {
      System.out.println("JmsEvent:: Error during unmarshalling " + ioe);
    } 
  }


  public byte[] getBytesWithoutPayloadReconstruct() throws JMSException {
    ByteArrayOutputStream baOutputStream = new ByteArrayOutputStream();
    DataOutputStream dout = new DataOutputStream(baOutputStream);
    
    try {
      /** The byte header identifying the message packet */
      dout.writeByte(Protocol.JMS_EVENT);
      dout.writeUTF(topic);
      
      dout.writeByte(messageType);
      
      /** Write the length of the payload */
      dout.writeInt(payloadBytes.length);
      /** Write the Payload */
      dout.write(payloadBytes);
      dout.writeInt(deliveryMode);
      dout.writeInt(priority);
      dout.writeLong(timeToLive);
      dout.flush();
    } catch (IOException ioe) {
      System.out.println("JMSEvent: Error in marshalling JMSEvent Stream" 
			 + ioe);
      throw new JMSException("JMSEvent: Error in marshalling JMSEvent Stream" 
			     + ioe);
    }
    
    return baOutputStream.toByteArray();
  }
  
  public byte[] getBytes() throws JMSException {
    ByteArrayOutputStream baOutputStream = new ByteArrayOutputStream(300);
    DataOutputStream dout = new DataOutputStream(baOutputStream);
    
    try {
      /** The byte header identifying the message packet */
      dout.writeByte(Protocol.JMS_EVENT);
      
      dout.writeUTF(topic);
      
      if (message.getJMSType().equals("TextMessage")) {
	/** write the type */
	dout.writeByte(JmsProtocolConstants.TextMessage_Type);
	payloadBytes =  ((JmsTextMessage)message).getBytes();
	/** Write the length of the payload */
	dout.writeInt(payloadBytes.length);
	/** Write the Payload */
	dout.write(payloadBytes);
      } else if (message.getJMSType().equals("BytesMessage")) {
	/** write the type */
	dout.writeByte(JmsProtocolConstants.BytesMessage_Type);
	payloadBytes = ((JmsBytesMessage)message).getBytes();
	/** Write the length of the payload */
	dout.writeInt(payloadBytes.length);
	/** Write the Payload */
	dout.write(payloadBytes);
      } else if (message.getJMSType().equals("ObjectMessage")) {
	/** write the type */
	dout.writeByte(JmsProtocolConstants.ObjectMessage_Type);
	payloadBytes = ((JmsObjectMessage)message).getBytes();
	/** Write the length of the payload */
	dout.writeInt(payloadBytes.length);
	/** Write the Payload */
	dout.write(payloadBytes);
      } else if (message.getJMSType().equals("StreamMessage")) {
	/** write the type */
	dout.writeByte(JmsProtocolConstants.StreamMessage_Type);
	payloadBytes = ((JmsStreamMessage)message).getBytes();
	/** Write the length of the payload */
	dout.writeInt(payloadBytes.length);
	/** Write the Payload */
	dout.write(payloadBytes);
      } else if (message.getJMSType().equals("MapMessage")) {
	/** write the type */
	dout.writeByte(JmsProtocolConstants.MapMessage_Type);
	payloadBytes = ((JmsMapMessage)message).getBytes();
	/** Write the length of the payload */
	dout.writeInt(payloadBytes.length);
	/** Write the Payload */
	dout.write(payloadBytes);
      } else {
	System.out.println("\n" + "**************No idea how to serialize " +
			   " this message type ******************* " + "\n");
	return null; /** A null value indicates the serialization has failed*/
      }
      
      dout.writeInt(deliveryMode);
      dout.writeInt(priority);
      dout.writeLong(timeToLive);
    } catch (IOException ioe) {
      System.out.println("JMSEvent: Error in marshalling JMSEvent Stream" 
			 + ioe);
    }
    return baOutputStream.toByteArray();
  }

  public String getTopic() {
    return topic;
  }
  
  public Message getMessage() {
    return message;
  }

  public int getDeliveryMode() {
    return deliveryMode;
  }

  public int getPriority() {
    return priority;
  }

  public long getTimeToLive() {
    return timeToLive;
  }
  
  public String toString() {
    String stringRep = "Topic = " + topic.toString() + " PayLoad " + 
      message.toString() + " DeliveryMode =" + deliveryMode + 
      " Priority ="  + priority + " TimeToLive =" + timeToLive;
    
    return stringRep;
  }

  public static void main(String[] args) {
    try {
      String topic = "Scores/NBA/Team/Wizards/Players/Jordan";
      JmsTextMessage jmsText = new JmsTextMessage();
      jmsText.setText("Jordan rules ...");
      JmsEvent jmsEvent= new JmsEvent(topic, jmsText, 0, 3, 10);
      System.out.print("Marshalling ....");
      byte[] marshalledBytes = jmsEvent.getBytes();
      System.out.print("Completed" + "\n");
      System.out.print("Unmarshalling ....");
      System.out.println(new JmsEvent(marshalledBytes));
      System.out.println("Success");

      Hashtable testTable = new Hashtable();
      testTable.put("Lord of ", "THE RINGS");

      JmsObjectMessage jmsObjMsg = new JmsObjectMessage();
      jmsObjMsg.setObject(testTable);
      
      jmsEvent= new JmsEvent(topic, jmsObjMsg, 0, 3, 10);
      System.out.print("Marshalling ....");
      marshalledBytes = jmsEvent.getBytes();
      System.out.print("Completed" + "\n");
      System.out.print("Unmarshalling ....");
      System.out.println(new JmsEvent(marshalledBytes));
      System.out.println("Success");
      
    } catch (JMSException jmsEx) {
      System.out.println(jmsEx);
    }
  }
}



